package net.bw.realtime.warehouse.dwd;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bw.realtime.warehouse.dwd
 * @Author: zhangleqing
 * @CreateTime: 2025-08-25  20:31
 * @Description: TODO
 * @Version: 1.0
 */
public class Room {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

/*
CREATE TABLE ods_room(
    `id` INT,
    `houseId` INT,
    `roomCode` STRING,
    `roomName` STRING,
    `peopleNum` INT,
    `remark` STRING,
    PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc', -- 使用 MySQL CDC 连接器
    'hostname' = 'node101', -- MySQL 服务器主机名
    'port' = '3306', -- MySQL 服务端口
    'username' = 'root', -- 数据库用户名
    'password' = '123456', -- 数据库密码
    'database-name' = 'sca', -- 数据库名称
    'table-name' = 'room', -- 源表名称
    'connect.timeout' = '60s' -- 连接超时时间（60秒）
)
*/
        // 1.读取MySQL表数据
        tEnv.executeSql(
                "CREATE TABLE ods_room(\n" +
                        "    `id` INT,\n" +
                        "    `houseId` INT,\n" +
                        "    `roomCode` STRING,\n" +
                        "    `roomName` STRING,\n" +
                        "    `peopleNum` INT,\n" +
                        "    `remark` STRING,\n" +
                        "    PRIMARY KEY (`id`) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "    'connector' = 'mysql-cdc', -- 使用 MySQL CDC 连接器\n" +
                        "    'hostname' = 'node101', -- MySQL 服务器主机名\n" +
                        "    'port' = '3306', -- MySQL 服务端口\n" +
                        "    'username' = 'root', -- 数据库用户名\n" +
                        "    'password' = '123456', -- 数据库密码\n" +
                        "    'database-name' = 'sca', -- 数据库名称\n" +
                        "    'table-name' = 'room', -- 源表名称\n" +
                        "    'connect.timeout' = '60s' -- 连接超时时间（60秒）\n" +
                        ")"
        );

/*
create table ods_room_clean(
    `id` INT,
    `houseId` INT,
    `roomCode` STRING,
    `roomName` STRING,
    `peopleNum` INT,
    `remark` STRING,
    PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'ods_room_clean',
    'properties.bootstrap.servers' = 'node101:9092',
    'key.format' = 'json',
    'value.format' = 'json',
    'sink.delivery-guarantee' = 'exactly-once',
    'properties.transaction.timeout.ms' = '900000',
    'sink.parallelism' = '3'
)
*/
        // 2.创建Kafka数据源表
        tEnv.executeSql(
                "create table ods_room_clean(\n" +
                        "    `id` INT,\n" +
                        "    `houseId` INT,\n" +
                        "    `roomCode` STRING,\n" +
                        "    `roomName` STRING,\n" +
                        "    `peopleNum` INT,\n" +
                        "    `remark` STRING,\n" +
                        "    PRIMARY KEY (`id`) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "    'connector' = 'upsert-kafka',\n" +
                        "    'topic' = 'ods_room_clean',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'key.format' = 'json',\n" +
                        "    'value.format' = 'json',\n" +
                        "    'sink.delivery-guarantee' = 'exactly-once',\n" +
                        "    'properties.transaction.timeout.ms' = '900000',\n" +
                        "    'sink.parallelism' = '3'\n" +
                        ")"
        );

/*
insert into ods_room_clean
select
    id,
    houseId,
    roomCode,
    roomName,
    peopleNum,
    remark
from ods_room
WHERE id IS NOT NULL
  AND houseId IS NOT NULL
*/
        // 3.插入数据
        tEnv.executeSql(
                "insert into ods_room_clean\n" +
                        "select\n" +
                        "    id,\n" +
                        "    houseId,\n" +
                        "    roomCode,\n" +
                        "    roomName,\n" +
                        "    peopleNum,\n" +
                        "    remark\n" +
                        "from ods_room\n" +
                        "WHERE id IS NOT NULL\n" +
                        "  AND houseId IS NOT NULL"
        );

        env.execute("Room");
    }
}
